package com.atguigu.timer;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Created by RUI on 2021/6/16 9:08
 *
 * 监控水位传感器的水位值，如果水位值在5s之内(event time)连续上升，则报警。
 */
public class Timer1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        env.socketTextStream("hadoop162",9999)
                .map(value -> {
                    String[] split = value.split(",");
                    return new WaterSensor(split[0],Long.parseLong(split[1])*1000,Integer.valueOf(split[2]));
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                            @Override
                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
                )
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {

                    WaterSensor ws;
                    boolean isFirstVc = true;

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                        if (isFirstVc){
                            System.out.println(ctx.timestamp());
                            ctx.timerService().registerEventTimeTimer(ctx.timestamp()+5000);
                            isFirstVc = false;
                        }else {

                            if (value.getVc() - ws.getVc() <= 0) {
                                System.out.println("水位下降或不变");
                                ctx.timerService().deleteEventTimeTimer(ws.getTs()+5000);

                                ctx.timerService().registerEventTimeTimer(ctx.timestamp()+5000);
                            }else {
                                System.out.println("水位上升");
                            }
                        }

                        ws = value;
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect(ctx.getCurrentKey()+": 连续5s水位上升，发出预警");
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
